package concurrentwrite.callback.run;

//import concurrentwrite.callback.run.multicluster.MainMultiCluster;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.MutationResult;
import me.prettyprint.hector.api.mutation.Mutator;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.lang.reflect.InvocationTargetException;
import java.util.concurrent.Callable;

public class WriteLogByFileTask implements Callable {
    private Main callBackTest;
    private int seq;
    private String filePath;
    private static final String columnFamilyName = "AdLog";
    private Cluster cluster;
    private final String adsData = "AdsData";

    public Cluster getCluster() {
        return cluster;
    }

    public void setCluster(Cluster cluster) {
        this.cluster = cluster;
    }

    public String getFilePath() {
        return filePath;
    }

    public void setFilePath(String filePath) {
        this.filePath = filePath;
    }

    public WriteLogByFileTask() {
    }

    public WriteLogByFileTask(int i) {
        seq = i;
    }

    public Object call() {
        String str = "";
        long begTest = System.currentTimeMillis();
        System.out.println("start - Task " + seq);
        try {
            int max_rows_a_batch = 10000;
            Keyspace keyspaceOperator = HFactory.createKeyspace(adsData, cluster);
            Mutator<String> mutator = HFactory.createMutator(keyspaceOperator, StringSerializer.get());

            long startTick = System.currentTimeMillis();
            BufferedReader reader = new BufferedReader(new FileReader(DataQueue.folderPath + File.separator + filePath));
            String thisLine;
            int countRows = 0;
            while ((thisLine = reader.readLine()) != null) {
                addToBatch(mutator, thisLine);
                countRows++;
                if (countRows == max_rows_a_batch) {
                    MutationResult result = mutator.execute();
                    System.out.println(((double) result.getExecutionTimeMicro() / (double) (1000 * 1000)) + " host:" + result.getHostUsed().getHost());
                    countRows = 0;
                    //break;
                }
            }
            MutationResult result = mutator.execute();
            System.out.println(((double) result.getExecutionTimeMicro() / (double) (1000 * 1000)) + " host:" + result.getHostUsed().getHost());
//
            System.out.println("file " + filePath + (System.currentTimeMillis() - begTest));
            reader.close();

        } catch (Exception e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }

        callBackTest.callBack(seq);

        Double secs = new Double((new java.util.Date().getTime() - begTest) * 0.001);
        System.out.println("task -" + seq + " took " + secs + " secs");
        return null;
    }

    public void setCaller(Main callBackTest) {
        this.callBackTest = callBackTest;
    }

    public Main getCaller() {
        return callBackTest;
    }

    private static void addToBatch(Mutator<String> mutator, String aLine) throws InvocationTargetException, NoSuchMethodException, IllegalAccessException {
        String[] stringArr = aLine.split("\t");

        // row key
        String key = String.valueOf(System.nanoTime());

        // add
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("time_create", stringArr[0]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("browser_code", stringArr[1]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("browser_name", stringArr[2]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("browser_ver", stringArr[3]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("os_name", stringArr[4]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("os_code", stringArr[5]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("os_ver", stringArr[6]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("bannerId", stringArr[7]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("ip", stringArr[8]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("domain", stringArr[9]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("path", stringArr[10]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("clickorView", stringArr[11]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("cookie_create", stringArr[12]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("viewCount", stringArr[13]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("guid", stringArr[14]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("day", stringArr[15]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("month", stringArr[16]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("year", stringArr[17]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("hour", stringArr[18]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("minute", stringArr[19]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("isNewIndex", stringArr[20]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("geographic", stringArr[21]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("zoneID", stringArr[22]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("campain", stringArr[23]));
        mutator.addInsertion(key, columnFamilyName, HFactory.createStringColumn("channelID", stringArr[24]));


        //HColumn<String, String> column = HFactory.createStringColumn(colName, value);
        return;
    }
}